package cn.itcast.flink.base;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author itcast
 * Date 2021/7/27 15:25
 * 对流中的数据按照奇数和偶数进行分流，并获取分流后的数据
 */
public class SideOutputTransformation {
    public static void main(String[] args) throws Exception {
        /**
         * 1.生成流执行环境，获取本地数据
         * 2.定义两个侧输出流，一个用于保存偶数，一个用于保存奇数
         * 3.切分数据流
         * 4.打印输出偶数数据流和奇数数据流
         * 5.执行环境
         */
        //1.生成流执行环境，获取本地数据
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Long> source = env.fromSequence(1, 10);
        //2.定义两个侧输出流，一个用于保存偶数，一个用于保存奇数
        OutputTag odd = new OutputTag("odd", TypeInformation.of(Long.class));
        OutputTag even = new OutputTag("even", TypeInformation.of(Long.class));
        //3.切分数据流
        SingleOutputStreamOperator<Long> result = source.process(new ProcessFunction<Long, Long>() {
            @Override
            public void processElement(Long value, Context ctx, Collector<Long> out) throws Exception {
                //判断满足被5整除的数字能够打印输出，否则除了5之外的偶数放到 even侧输出tag ，奇数放在 odd 侧输出 tag
                if (value % 5 == 0) {
                    out.collect(value);
                }
                if (value % 2 == 0) {
                    ctx.output(even, value);
                } else {
                    ctx.output(odd, value);
                }
            }
        });
        //4.打印输出偶数数据流和奇数数据流
        result.print("能够被5整除的");
        result.getSideOutput(odd).print("奇数");
        result.getSideOutput(even).print("偶数");
        //5.执行环境
        env.execute();
    }
}
